package com.atguigu.flink0922.chapter11;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/12 11:09
 */
public class Flink12_Function_Scalar {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
        DataStreamSource<String> stream = env.fromElements("hello", "atguigu", "Hello");
    
        Table table = tEnv.fromDataStream(stream, $("word"));
        
        // 1. table api 使用方式1: 不注册直接 inline 使用
        table.select(call(ToUpperCase.class, $("word")).as("word_upper")).execute().print();
        // 2. table api 使用方式2: 注册后使用
        // 2.1 注册函数
        tEnv.createTemporaryFunction("toUpper", ToUpperCase.class);
        // 2.2 使用函数
        table.select(call("toUpper", $("word")).as("word_upper")).execute().print();
        
        
        
    }
    
    public static class ToUpperCase extends ScalarFunction {
        public String eval(String s){
            return s.toUpperCase();
        }
    }
    
    
}



